feat(ssestream): fix event decoding for data-only messages and [DONE] handling #261
Conversation
… handling Fixed the Stream[T].Next() method to properly handle data events that don't have an explicit event: field. Previously, when the decoder encountered data without an explicit event type, it would skip processing entirely. Now, when eventType is empty, the decoder attempts to unmarshal the data directly as JSON. Signed-off-by: yuguorui <[email protected]>
There was a problem hiding this comment.
Pull request overview
This PR fixes the SSE stream decoder to properly handle data-only messages (without explicit event types) and [DONE] sentinel messages. Previously, messages without an event type field were skipped entirely, causing data loss.
Key changes:
- Added explicit handling for [DONE] messages to gracefully terminate the stream
- Introduced fallback processing for events without an explicit type field by attempting direct JSON unmarshaling
- Optimized by caching event type and data to avoid repeated decoder calls
Comments suppressed due to low confidence (1)
packages/ssestream/ssestream.go:190
- The JSON unmarshaling and assignment logic is duplicated three times. Consider extracting this into a helper method or consolidating the cases to reduce code duplication and improve maintainability.
if eventType == "" {
// For Anthropic's SSE format, data events without explicit type should still be processed
var nxt T
s.err = json.Unmarshal(eventData, &nxt)
if s.err != nil {
return false
}
s.cur = nxt
return true
}
switch eventType {
case "completion":
var nxt T
s.err = json.Unmarshal(eventData, &nxt)
if s.err != nil {
return false
}
s.cur = nxt
return true
case "message_start", "message_delta", "message_stop", "content_block_start", "content_block_delta", "content_block_stop":
var nxt T
s.err = json.Unmarshal(eventData, &nxt)
if s.err != nil {
return false
}
s.cur = nxt
return true
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| eventData := s.decoder.Event().Data | ||
|
|
||
| // Check for [DONE] message which indicates end of stream | ||
| if string(eventData) == "[DONE]\n" || string(eventData) == "[DONE]" { |
There was a problem hiding this comment.
Converting byte slices to strings for comparison is inefficient. Use bytes.Equal() or bytes.HasPrefix() instead to avoid unnecessary allocations. For example: bytes.Equal(eventData, []byte("[DONE]")) or handle the newline separately.
| if string(eventData) == "[DONE]\n" || string(eventData) == "[DONE]" { | |
| if bytes.Equal(eventData, []byte("[DONE]\n")) || bytes.Equal(eventData, []byte("[DONE]")) { |
|
Functional: 13:24:29.942280 ssestream.go:70: bytes: event: content_block_start 13:24:30.178601 ssestream.go:70: bytes: event: content_block_delta Problematic (which is missing the |
1 participant
Fixed the Stream[T].Next() method to properly handle data events that don't have an explicit event: field. Previously, when the decoder encountered data without an explicit event type, it would skip processing entirely. Now, when eventType is empty, the decoder attempts to unmarshal the data directly as JSON.